<?php


namespace app\mqtt\controller;

use think\Exception;
use think\worker\Server;
use Workerman\Connection\AsyncTcpConnection;

class Mqtt extends Server
{
    protected $socket = 'tcp://0.0.0.0:1111';
    protected $protocol="tcp";
    protected $option = [
        'name' => 'msg_server',
        'count' => 50
    ];
    private $mqtt =null ;

    // onWorkerStart
    function onWorkerStart ($worker) {
        $mqtt= new \Workerman\Mqtt\Client('mqtt://47.99.158.37:1883');

        $mqtt->onConnect = function($mqtt) {
            $mqtt->subscribe(array('curl/+'=>2,'read_msg'=>2));
        };
        $mqtt->onMessage =function ($topic, $content) {
            $topic=explode("/",$topic);
            if($topic[0]=="curl"){
                $this->curl($content);
            }elseif ($topic=='read_msg'){
                $this->read_msg($content);
            }
        };
        $this->mqtt=$mqtt;
        $mqtt->connect();
        echo 'mqtt start';
    }

    /**
     * $data内容格式
     * {
     * 	"url": "http://generate.51xmm.net/",//地址（必须）
     * 	"args": {                           //参数（可空）
     * 		"a": "fasdf"
     * 	},
     * 	"from_account_id": "100",           //来源（可空）
     * 	"to_account_id": "20"               //发送给谁（可空）
     * }
     */

    function curl($data) {
        // 链接到MsgServer，如果是本机就是127.0.0.1，如果是集群就是lvs的ip
        $task_connection = new AsyncTcpConnection('tcp://127.0.0.1:1111');
        // 任务及参数数据

        // 发送数据
        $task_connection->send($data);
        // 异步获得$this->>onMessage结果
        $task_connection->onMessage = function($task_connection, $result)
        {
//            logs("get".time(),json_decode($result));

            $task_connection->close();
        };
        // 执行异步连接
        $task_connection->connect();
    }
    function read_msg($msg_id){
        if($msg_id){
            $accept_time = time();
            Db::name("account_msg")
                ->where('id',$msg_id)
                ->update(['is_accepted' => 1,'accept_time'=>$accept_time,'is_sended'=>1]);
        }
    }
    /**
     * $data内容格式
     * {
     * 	"url": "http://generate.51xmm.net/",//地址（必须）
     * 	"args": {                           //参数（可空）
     * 		"a": "fasdf"
     * 	},
     * 	"from_account_id": "100",           //来源（可空）
     * 	"to_account_id": "20"               //发送给谁（可空）
     * }
     */
    // onMessage
    function onMessage($connection, $data) {
        $data=json_decode($data,true);
        if(!isset($data['args'])){
            $data['args']=null;
        }
        try{
            if(isset($data['method'])&&$data['method']=="post"){
                $result = \tools\curl\Curl::post($data['url'], $data['args']);
            }else{
                $result = \tools\curl\Curl::get($data['url'], $data['args']);
            }
        }catch (Exception $e){
            logs("Exception".time(),$e->getTraceAsString());
            $result['status'] = false;
            $result['msg'] = '数据异常';
            $result['error'] = $e;
            $result=json_encode($result);
        }
        if(isset($data['to_account_id'])){
            $topic="curl_res/{$data['to_account_type']}_{$data['to_account_id']}";
            $message=json_encode(array('curl_data'=>$data,"result"=>$result));
            $options['qos']=2;
            $options['dup']=true;
            $this->mqtt->publish($topic,$message,$options);
        }
//        dump($result);
        $connection->send($result);
    }
}